/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.gaia.translation.wrappers.streaming.io;



import com.bff.gaia.unified.sdk.coders.CoderException;

import com.bff.gaia.unified.sdk.testing.TestStream;

import com.bff.gaia.unified.sdk.transforms.SerializableFunction;

import com.bff.gaia.unified.sdk.transforms.windowing.GlobalWindow;

import com.bff.gaia.unified.sdk.transforms.windowing.PaneInfo;

import com.bff.gaia.unified.sdk.util.WindowedValue;

import com.bff.gaia.unified.sdk.values.TimestampedValue;

import com.bff.gaia.streaming.api.functions.source.RichSourceFunction;

import com.bff.gaia.streaming.api.watermark.Watermark;

import org.joda.time.Instant;



import java.util.List;



/** Gaia source for executing {@link TestStream}. */

public class TestStreamSource<T> extends RichSourceFunction<WindowedValue<T>> {



  private final SerializableFunction<byte[], TestStream<T>> testStreamDecoder;

  private final byte[] payload;



  private volatile boolean isRunning = true;



  public TestStreamSource(

      SerializableFunction<byte[], TestStream<T>> testStreamDecoder, byte[] payload) {

    this.testStreamDecoder = testStreamDecoder;

    this.payload = payload;

  }



  @Override

  public void run(SourceContext<WindowedValue<T>> ctx) throws CoderException {

    TestStream<T> testStream = testStreamDecoder.apply(payload);

    List<TestStream.Event<T>> events = testStream.getEvents();



    for (int eventId = 0; isRunning && eventId < events.size(); eventId++) {

      TestStream.Event<T> event = events.get(eventId);



      synchronized (ctx.getCheckpointLock()) {

        if (event instanceof TestStream.ElementEvent) {

          for (TimestampedValue<T> element : ((TestStream.ElementEvent<T>) event).getElements()) {

            Instant timestamp = element.getTimestamp();

            WindowedValue<T> value =

                WindowedValue.of(

                    element.getValue(), timestamp, GlobalWindow.INSTANCE, PaneInfo.NO_FIRING);

            ctx.collectWithTimestamp(value, timestamp.getMillis());

          }

        } else if (event instanceof TestStream.WatermarkEvent) {

          long millis = ((TestStream.WatermarkEvent<T>) event).getWatermark().getMillis();

          ctx.emitWatermark(new Watermark(millis));

        } else if (event instanceof TestStream.ProcessingTimeEvent) {

          // There seems to be no clean way to implement this

          throw new UnsupportedOperationException(

              "Advancing Processing time is not supported by the Gaia Runner.");

        } else {

          throw new IllegalStateException("Unknown event type " + event);

        }

      }

    }

  }



  @Override

  public void cancel() {

    this.isRunning = false;

  }

}